package com.hery.services;

import com.alibaba.fastjson.JSONObject;
import com.googlecode.aviator.AviatorEvaluator;
import com.googlecode.aviator.AviatorEvaluatorInstance;
import com.googlecode.aviator.Options;
import com.hery.pojo.AppType;
import com.hery.pojo.Rule;
import com.hery.utils.BeanToMap;
import com.hery.utils.MysqlConnectionUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;

import java.io.Serializable;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * 管理规则的服务，负责规则的更新、维护
 */
@Slf4j
@Data
public class RulesService implements Serializable{

    private AppType appType;
    private String mysqlRuleSql;
    private String mysqlUrl;

    private String url;
    private String user;
    private String pswd;


    /**
     * 保存全量的规则
     */
    private HashMap<Integer, Rule> rules = new HashMap<>();

    public RulesService(Configuration config) {
        ConfigOption<AppType> appTypeOption = ConfigOptions.key("app.type").enumType(AppType.class).noDefaultValue();
        ConfigOption<String> ruleSqlOption = ConfigOptions.key("rule.sql").stringType().noDefaultValue();
        ConfigOption<String> uelOption = ConfigOptions.key("mysql.url").stringType().noDefaultValue();
        appType =  config.get(appTypeOption);
        mysqlRuleSql = config.get(ruleSqlOption);
        mysqlUrl = config.get(uelOption);

        ConfigOption<String> urlOption = ConfigOptions.key("mysql.url").stringType().noDefaultValue();
        ConfigOption<String> userOption = ConfigOptions.key("user").stringType().noDefaultValue();
        ConfigOption<String> pswdOption = ConfigOptions.key("pswd").stringType().noDefaultValue();
        url = config.get(urlOption);
        user = config.get(userOption);
        pswd = config.get(pswdOption);
    }

    /**
     * 全量同步 mysql 的规则到{@link #rules} 适用于：
     *      1、增量同步第一次启动时
     *      2、全量同步
     */
    public void allSyncRule() {

        try {
            MysqlConnectionUtil connUtil = new MysqlConnectionUtil(url, user, pswd);
            ArrayList<Rule> rules = connUtil.onceSelectToJavaBean(mysqlRuleSql, Rule.class);
            for (Rule rule : rules) {
                if(isValid(rule)) {
                    this.rules.put(rule.getId(), rule);
                }
            }
        } catch (SQLException e) {
            log.error("all sync mysql rule error, url: {}, sql: {}", mysqlUrl, mysqlRuleSql);
            e.printStackTrace();
        }
    }

    /**
     * 增量同步规则到 this.rules
     * @param value : binlog 内容
     */
    public void incrementSyncRule(String value) {

        Rule rule = RulesService.toRule(value);

        if(isValid(rule)) {
            rules.put(rule.getId(), rule);
        }

    }

    /**
     * 增量同步规则到 this.rules
     * @param rule : Rule 对象
     */
    public void incrementSyncRule(Rule rule) {
        // 更新规则库
        if(isValid(rule)) {
            rules.put(rule.getId(), rule);
        }
    }

    private boolean isValid(Rule rule) {
        return rule.getIsValid() == 1 && appType.name().equalsIgnoreCase(rule.getAppType());
    }


    private static Rule toRule(String binlogJson){
        try {
            // 获取更新后的变化内容
            String after = JSONObject.parseObject(binlogJson).getJSONObject("after").toJSONString();

            // 解析成 Rule 类型
            return JSONObject.parseObject(after, Rule.class);

        } catch (Exception e) {
            e.printStackTrace();
            log.error("parse MySQL rules error：{}", binlogJson);
        }
        return new Rule(0);

    }


    public List<Rule> mathRules(String json) {
        List<Rule> res = new ArrayList<>();

        Map<String, Object> map = BeanToMap.jsonFlatten(json);

        for (Map.Entry<Integer, Rule> entry : this.getRules().entrySet()) {
            try {
                Rule rule = entry.getValue();

                AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
                instance.setOption(Options.ENABLE_PROPERTY_SYNTAX_SUGAR, false);

                Object execute = AviatorEvaluator.execute(rule.getExp(), map);
                if (execute instanceof Boolean) {
                    boolean match = (boolean) execute;
                    if(match){
                        // 触发告警
                        log.info("Send Alert Data to AlertManager: | {} | <--> | {} |", rule, json);
                        res.add(rule);
                    }else{
                        log.debug("No Matched： | {} | -- | {} |", rule, json);
                    }
                }else{
                    log.error("rule set error, please modify rule:{}", rule);
                }
            } catch (Exception e) {
                log.error("AviatorEvaluator regular expression match error:| {} | -- | {} |",  this.getRules(), json);
                e.printStackTrace();
            }

        }
        return res;
    }
}
